/*
 * Copyright (c) Facebook, Inc. and its affiliates.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#pragma once

#include "velox/common/base/BitUtil.h"
#include "velox/common/base/Exceptions.h"
#include "DeltaBpDecoder.h"
#include "DeltaLengthBaDecoder.h"

namespace facebook::velox::parquet {

// DeltaBpDecoder is adapted from Apache Arrow:
// https://github.com/apache/arrow/blob/apache-arrow-12.0.0/cpp/src/parquet/encoding.cc#LL2357C18-L2586C3
//template <typename T>
class DeltaBaDecoder {
 public:
  explicit DeltaBaDecoder(const char* start, int32_t encode_size) 
	               :bufferStart_(start),
		       page_size_(encode_size),
		       prefix_len_decoder_(start){
    std::cout << "@@ init deltaBaDecoder \n";
    uint64_t len = prefix_len_decoder_.ValidValuesCount();
    prefix_len_decoder_.setReadNums(len);
    for(int i = 0; i < len; i++){
      int64_t v = prefix_len_decoder_.getReadLong();
      buffered_prefix_length_.push_back(v);
    }
    numValidValues_ = len;
    prefixLenOffset_ = 0;

    uint32_t byteOffset= 0;
    uint32_t bitOffset = 0;
    prefix_len_decoder_.getOffset(byteOffset, bitOffset);
    const char* prefix_len_addr = prefix_len_decoder_.getAddress();
    uint32_t byte_offset = calcByteOffset(bufferStart_, prefix_len_addr);

    const char* suffix_addr = start + byte_offset;
    int32_t suffix_len = page_size_ - byte_offset;
    suffix_decoder_.SetDecode(suffix_addr, suffix_len,numValidValues_);

    std::cout << "@@ first byte_offset:" << byte_offset <<"\n";
  }

  void skip(uint64_t numValues) {
    skip<false>(numValues, 0, nullptr);
  }

  template <bool hasNulls>
  inline void skip(int32_t numValues, int32_t current, const uint64_t* nulls) {
    if (hasNulls) {
      numValues = bits::countNonNulls(nulls, current, current + numValues);
    }
    for (int32_t i = 0; i < numValues; ++i) {
      readString();
    }
  }

  template <bool hasNulls, typename Visitor>
  void readWithVisitor(const uint64_t* nulls, Visitor visitor) {
    std::cout << "@@ readWithVisitor deltaBaDecoder \n";
    int32_t current = visitor.start();
    skip<hasNulls>(current, 0, nulls);
    int32_t toSkip;
    bool atEnd = false;
    const bool allowNulls = hasNulls && visitor.allowNulls();
    for (;;) {
      if (hasNulls && allowNulls && bits::isBitNull(nulls, current)) {
        toSkip = visitor.processNull(atEnd);
      } else {
        if (hasNulls && !allowNulls) {
          toSkip = visitor.checkAndSkipNulls(nulls, current, atEnd);
          if (!Visitor::dense) {
            skip<false>(toSkip, current, nullptr);
          }
          if (atEnd) {
            return;
          }
        }

        // We are at a non-null value on a row to visit.
        toSkip = visitor.process(readString(), atEnd);
      }
      ++current;
      if (toSkip) {
        skip<hasNulls>(toSkip, current, nulls);
        current += toSkip;
      }
      if (atEnd) {
        return;
      }
    }
  }

 private:

  void buildInternal(uint64_t prefix_len_value, std::string_view* suffix_v,std::string_view* prefix, uint8_t** data_ptr){
    if(prefix_len_value > prefix->length()){
      VELOX_FAIL("negative prefix length in DELTA_BYTE_ARRAY");
    }

    if(prefix_len_value == 0){
      *prefix = *suffix_v;
      return ;
    }
 
    if(suffix_v->size() == 0){
      *prefix = prefix->substr(0,prefix_len_value);
      *suffix_v = *prefix;
      return;
    }

    //1.cpy the prefix
    memcpy(*data_ptr, prefix->data(), prefix_len_value);
    //2. cpy the suffix
    memcpy(*data_ptr + prefix_len_value, suffix_v->data(), suffix_v->size());
    //3. Point suffix to the decoded string
    std::string_view value((char*)(*data_ptr), (uint64_t)(prefix_len_value+suffix_v->size()));
    *prefix = value;
  }

  folly::StringPiece  readString() {
    std::cout << "@@ readString deltaBaDecoder \n";
    //std::string value ;
    
    std::string_view value = suffix_decoder_.getString();
    int64_t prefix_v = buffered_prefix_length_[prefixLenOffset_];
    if(prefix_v < 0){
      VELOX_FAIL("negative prefix length in DELTA_BYTE_ARRAY");
    }

    int64_t data_size = 0;
    if(prefix_v != 0){
      data_size += prefix_v;
      data_size += value.size();
    }
    if(value.size() == 0){

    }
    //int64_t data_size = prefix_v;
    //data_size += value.size();
    buffered_data_.resize(data_size);
    uint8_t* data = buffered_data_.data();
    std::string_view prefix{lastValue_};

    buildInternal(prefix_v, &value, &prefix, &data);

    prefixLenOffset_++;
    numValidValues_--;

    lastValue_ = std::string(prefix);

    if(numValidValues_ == 0){
      lastValueInPreviousPage_ = lastValue_;
    }

    return folly::StringPiece(value);
  }

  uint32_t calcByteOffset(const char* org, const char* cur){
    return uint32_t(cur - org);
  }

  const char* bufferStart_;
  int32_t page_size_;

  std::string lastValue_;
  std::string lastValueInPreviousPage_;
  int numValidValues_{0};
  uint32_t prefixLenOffset_{0};
  std::vector<int64_t> buffered_prefix_length_;
  std::vector<uint8_t> buffered_data_;

  DeltaBpDecoder prefix_len_decoder_;
  DeltaLengthBaDecoder suffix_decoder_;
};

} // namespace facebook::velox::parquet
